/*
 * Copyright (c) 2007 The Trustees of Indiana University. All rights reserved.
 * IU CSCI B534 Distributed Systems 2007-02-01 http://www.cs.indiana.edu/classes/b534
 */

package edu.indiana.cs.b534.torrent.impl;

import edu.indiana.cs.b534.torrent.MessageType;
import edu.indiana.cs.b534.torrent.StorageManager;
import edu.indiana.cs.b534.torrent.TorrentException;
import edu.indiana.cs.b534.torrent.Utils;
import edu.indiana.cs.b534.torrent.context.PeerDescription;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.logging.Logger;

/**
 * Manages the interactions with a single peer for a particular torrent
 */
public class PeerInstance implements Runnable {

    private InputStream in = null;
    private OutputStream out = null;
    //TODO set this to maxumum size of possible message, so it will never overflow
    //private byte[] buf = new byte[Constants.BLOCK_SIZE*2];

    private String connectingPeerId;
    private byte[] infoHash;
    private String remoteIP;
    private int remotePort;

    private PieceInfo currentPiece;
    private final StorageManager storageManager;

    Logger logger = Logger.getLogger(TorrentManager.TORRENT_MANAGER_NAME);

    /**
     * This is about my decisions.
     * amInterested - I expressed to the connecting peer that I am interested to it
     * amChoked - I expressed to the connecting peer that I choked it. Active voice ;)
     */
    private boolean amInterested = false, amChoked = true;

    /**
     * this is about peer's decisions
     * peerInterested - is it interested
     * peerChoked - has it choked me
     */
    private boolean peerInterested = false;
    //private boolean peerChoked = true;
    private PieceManager manager;
    private MessageUtil messageUtil;
    private int beginIndex = 0;
    private boolean busy = false;
    private final PeerDescription peerDescription;
    private final boolean isIncoming;

    public PeerInstance(InputStream in, OutputStream out, PieceManager pieceManager) throws TorrentException {
        this.in = in;
        this.out = out;
        byte[] buf = new byte[1024];
        Utils.readGivenNumberOfBytes(in, 4, buf);
        int msgLength = Utils.byteArrayToInt(buf, 0);

        logger.info("Connection with a message length " + msgLength + " received from a peer");

        Utils.readGivenNumberOfBytes(in, msgLength, buf);

        messageUtil = new MessageUtil(buf, pieceManager.getTorrentContext().getInfoHash());
        //TODO this does not send IP and port, what to do?
        this.remoteIP = null;
        this.remotePort = 0;
        this.infoHash = messageUtil.getInfoHash();

        manager = pieceManager;

        this.connectingPeerId = messageUtil.getPeerId();
        this.storageManager = manager.getTorrentContext().getStorageManager();

        PeerDescription temPeerDescription = manager.getTorrentContext().getPeerDescription(connectingPeerId);
        if (temPeerDescription == null) {
            temPeerDescription = new PeerDescription(connectingPeerId);
            manager.getTorrentContext().addPeerDescription(temPeerDescription);
        }
        peerDescription = temPeerDescription;

        logger.info("Handshake message received from " + connectingPeerId + ". Yet to send the response");
        this.isIncoming = true;
    }


    public PeerInstance(byte[] infoHash, String idAddress, int port, PieceManager manager, String peerID) throws TorrentException {
        try {
			this.infoHash = infoHash;
			remoteIP = idAddress;
			remotePort = port;
			this.manager = manager;
			messageUtil = new MessageUtil(infoHash, peerID);
			this.connectingPeerId = peerID;
			PeerDescription temPeerDescription = manager.getTorrentContext().getPeerDescription(connectingPeerId);
			if (temPeerDescription == null) {
			    temPeerDescription = new PeerDescription(connectingPeerId);
			    manager.getTorrentContext().addPeerDescription(temPeerDescription);
			}
			peerDescription = temPeerDescription;
			this.storageManager = manager.getTorrentContext().getStorageManager();
			//We do connection in constructer, so we know before starting the thread, the other socket it up or not
			Socket socket = new Socket(remoteIP, remotePort);

			logger.info("Connection established with peer in " + remoteIP + ":" + remotePort);
			this.isIncoming = false;
			in = socket.getInputStream();
			out = socket.getOutputStream();
		} catch (UnknownHostException e) {
			throw new TorrentException(e);
		} catch (IOException e) {
			throw new TorrentException(e);
		}
    }


    public String getID() {
//        return new StringBuffer().append(infoHash).append(":")
//                .append(remoteIP).append(":").append(remotePort).toString();
        return this.connectingPeerId;
    }

    //TODO we need sync here if we invoke same code from the server and peer instance
    public synchronized void handleMessage(MessageType type, byte[] payload) throws TorrentException {
        int pieceIndex;
        int begin;
        int length;
        switch (type) {
            case Choke:

                logger.info("Choke message received from " + peerDescription.getPeerId());
                // stop waiting for current piece. notify manager that piece canceled.
                manager.cancelPieceDownload(currentPiece);
                peerDescription.setChoked(true);
                busy = false;
                break;

            case UnChoke:
                logger.info("Un-Choke message received from " + peerDescription.getPeerId());

                // ask manager for a a piece to download from remote peer
                // 'request' the piece from peer and start timer
                //manager.getPieceToDownload(availableBitField)

                peerDescription.setChoked(false);
                break;

            case Interested:
                logger.info("Interested message received from " + peerDescription.getPeerId());

                if (manager.interestedMessageReceived(this)) {
                    amChoked = false;
                    this.sendUnChokeMessage();
                }
                peerInterested = true;

                break;

            case NotInterested:
                logger.info("Not Interested message received from " + peerDescription.getPeerId());

                // update interest of peer in pieces we have
                peerInterested = false;
                break;

            case Have:
                logger.info("Have message received from " + peerDescription.getPeerId());

                int haveIndex = Utils.byteArrayToInt(payload, 0);

                // update available bit field of pieces that remote peer has

                // TODO : It is better to put addPiece method in to the peer description, rather than
                // directly adding like this. The reason is, else things might go wrong when we get
                // duplicate have messages
                peerDescription.addPiece(haveIndex);

                break;

            case BitField:
                logger.info("BitField message received from " + peerDescription.getPeerId());

                // init available bit field of pieces that remote peer has
                // ask manager for a a piece to download from remote peer
                // if piece is to be downloaded, send interested message
                // if unchoked, 'request' the piece from peer and start timer
                boolean[] pieces = Utils.bitsToBooleanArray(payload);
                for (int i = 0; i < pieces.length; i++) {
                    if (pieces[i]) {
                        //TODO make sure repeated messages does not get here
                        peerDescription.addPiece(i);
                    }
                }

                break;

            case Request:
                logger.info("Request message received from " + peerDescription.getPeerId());

                // get the 'piece' from manager and send it to peer
                pieceIndex = Utils.byteArrayToInt(payload, 0);
                begin = Utils.byteArrayToInt(payload, 4);
                length = Utils.byteArrayToInt(payload, 8);

                if(pieceIndex != 0){
                    System.out.println("Need to debug");
                }

                if (!amChoked) {
                    // For me to be not choked, then by this time the corresponding peer must have sent me
                    // interested messages and all the handshakes are done. So I don't need to explicitly
                    // check them here
                	PieceInfo piceInfo = storageManager.getPieceInfo(pieceIndex);
                	if(piceInfo.isCompleted()){
                        byte[] piece = storageManager.getPiece(pieceIndex);
                        byte[] dataToBeSent = new byte[length];
                        //We do this to avoid array index out of bound on last block of the piece
                        int amoutOFDatabeCopied = Math.min(piece.length-begin, length);
                        System.arraycopy(piece, begin, dataToBeSent, 0, amoutOFDatabeCopied);
                        try {
                            this.sendMessage(messageUtil.buildPiece(pieceIndex, begin, dataToBeSent));
                        } catch (TorrentException e) {
                            logger.severe("Can not send piece to the peer. " + e);
                        }
                	}else{
                		throw new TorrentException("Something really went bad, I am seding data from a piece which is not completed");
                	}
                } else {
                    logger.info("I am choked to peer " + peerDescription.getPeerId() + ". Shutting down the connection, as I received a request message from it");
                    shutDownConnectionToPeer();
                }
                logger.info("Request processing Done");
                break;

            case Piece:
                logger.info("Piece message received from " + peerDescription.getPeerId());

                // receive a piece from peer and send it to manager.
                // ask manager for a a piece to download from remote peer
                // 'request' the piece from peer and start timer
                pieceIndex = Utils.byteArrayToInt(payload, 0);
                begin = Utils.byteArrayToInt(payload, 4);

                // save the piece
                storageManager.storeBlock(pieceIndex, begin, payload, 8);

                // check whether you have downloaded the whole thing, say the piece, that you are
                // supposed to download
                if (!storageManager.isPieceCompleted(pieceIndex)) {
                    // if not finish send the next request message to get the rest
                    synchronized (this) {
                        beginIndex = beginIndex + Constants.BLOCK_SIZE;
                    }
                    PieceInfo pieceInfo = storageManager.getPieceInfo(pieceIndex);
                    int amountLeftToDownload =  pieceInfo.getPieceLength()-beginIndex;
                    logger.info(amountLeftToDownload +"/"+pieceInfo.getPieceLength() + " left to be downloaded in piece "+pieceIndex);
                    int amountToDownLoadNext = Math.min(Constants.BLOCK_SIZE,amountLeftToDownload);
                    sendRequestMessage(pieceIndex, beginIndex, amountToDownLoadNext);
                } else {
                    manager.setPieceDownloaded(manager.getPieceInfo(pieceIndex), this.connectingPeerId);
                    busy = false;
                }
                logger.info("Piece Message "+ pieceIndex + " offset "+ begin + " is saved");
                break;
            case Cancel:
                logger.info("Cancel message received from " + peerDescription.getPeerId());

                // stop sending piece to peer
                pieceIndex = Utils.byteArrayToInt(payload, 0);
                begin = Utils.byteArrayToInt(payload, 4);
                length = Utils.byteArrayToInt(payload, 8);
                //TODO
                break;

            default:
                throw new TorrentException("Unknown message type");
        }
    }

    private void shutDownConnectionToPeer() {
        try {
            this.out.close();
            this.in.close();
            this.manager.getTorrentContext().removePeer(this.connectingPeerId);
        } catch (IOException e) {
            logger.info("Couldn't shutdown the connection to peer " + this.connectingPeerId);
        }
    }

    public void sendHaveMessage(PieceInfo piece) throws TorrentException {
        sendMessage(messageUtil.buildHave(piece.getIndex()));
        logger.info("Have message sent ..");
        // send 'have' message for piece to remote peer
    }

    public void sendUnChokeMessage() throws TorrentException {
        sendMessage(messageUtil.buildUnChoke());
        logger.info("Unchoke message sent ..");
    }

    public void sendInterested() throws TorrentException {
        sendMessage(messageUtil.buildInterested());
        peerDescription.setSentInterested(true);
        logger.info("Interested message sent ..");
    }

    public void run() {

        try {
            if (!isIncoming) {
                // send handshake to remote peer
                // handshake is first message: header, reserved bytes, infohash, peerid
                sendMessage(messageUtil.buildHandShake());

                logger.info("Handshake message sent .. ");

                // check if infohash is expected infohash
                // check if remote peerid is expected peerid
                readAndVerifyHandshake();

                logger.info("Handshake response received and verified ...");

                // if I have something to exchange with others, then let's send the bit field message
                boolean[] availability = storageManager.getPieceAvailability();
                if (availability != null) {
                    this.sendBitFieldMessage(availability);
                }

                // ok, now we are ready to connect with the peer. Let's send the interested message
                this.sendInterested();

            } else {
                //we have already verified the incoming handshake by this time
                sendMessage(messageUtil.buildHandShake());

                logger.info("Handshake response sent ...");

                // if I have something to exchange with others, then let's send the bit field message
                boolean[] availability = storageManager.getPieceAvailability();
                if (availability != null) {
                    this.sendBitFieldMessage(availability);
                }
            }

            // now we are done with shaking hands ;)

            //keep reading messages
            while (true) {
            	byte[] buf = new byte[Constants.BLOCK_SIZE*2];
                logger.info("Waiting for message from peer " + connectingPeerId);
                // get message length and content
                Utils.readGivenNumberOfBytes(in, 4, buf);
                int messageLength = Utils.byteArrayToInt(buf, 0);
                if (messageLength == 0) {
                    continue;
                }

                //get message type from reading first byte of message content
                // call handle message with type and rest of content as payload
                Utils.readGivenNumberOfBytes(in, messageLength, buf);
                byte messageType = buf[0];
                MessageType enumMessageType;
                switch (messageType) {
                    case 0:
                        enumMessageType = MessageType.Choke;
                        break;
                    case 1:
                        enumMessageType = MessageType.UnChoke;
                        break;
                    case 2:
                        enumMessageType = MessageType.Interested;
                        break;
                    case 3:
                        enumMessageType = MessageType.NotInterested;
                        break;
                    case 4:
                        enumMessageType = MessageType.Have;
                        break;
                    case 5:
                        enumMessageType = MessageType.BitField;
                        break;
                    case 6:
                        enumMessageType = MessageType.Request;
                        break;
                    case 7:
                        enumMessageType = MessageType.Piece;
                        break;
                    case 8:
                        enumMessageType = MessageType.Cancel;
                        break;
                    default:
                        throw new TorrentException("Unknown Message type " + messageType);
                }
                byte[] payload = new byte[messageLength - 1];
                System.arraycopy(buf, 1, payload, 0, messageLength - 1);
                handleMessage(enumMessageType, payload);
            }
        } catch (TorrentException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } 
    }

    private void sendBitFieldMessage(boolean[] pieceAvailability) throws TorrentException {
        sendMessage(messageUtil.buildBitField(pieceAvailability));
        logger.info("Bitfiled message sent ..");
    }


    private void sendMessage(byte[] content) throws TorrentException {
        // get content size and prepend to content
        // open socket to remoteIP:remotePort
        // send bytes to remote peer

        //TODO check does this output strream is alive or not, and reopen if it not alive
        try {
            if(content[0] >= 0 && content[0] <= 8 || content[0] == 19){
                out.write(Utils.intToByteArray(content.length));
                out.write(content);
                out.flush();
            }else{
            	throw new RuntimeException("Message does not starts with Message type, found  "+ content[0]);
            }
        } catch (IOException e) {
            throw new TorrentException(e);
        }
    }

    private void readAndVerifyHandshake() throws TorrentException {
    	byte[] buf = new byte[1024];
        Utils.readGivenNumberOfBytes(in, 4, buf);
        int messageLength = Utils.byteArrayToInt(buf, 0);

        assert messageLength == messageUtil.getCapacity();

        Utils.readGivenNumberOfBytes(in, messageUtil.getCapacity(), buf);
        //TODO verify peerID and info Hash
        //throw new UnsupportedOperationException();
    }


    private void sendRequestMessage(int pieceIndex, int begin, int length) throws TorrentException {
        byte[] bytes = messageUtil.buildRequest(pieceIndex, begin, length);
        this.sendMessage(bytes);
        logger.info("Request message sent to download piece " + pieceIndex + " offset "+ begin + " length "+ length);
    }


    public boolean isBusy() {
        return busy;
    }

    public void startDownload(int pieceIndex) throws TorrentException {
        if (busy) {
            throw new TorrentException("Trying to start new piece download when peer instance "+ remoteIP + ":"+remotePort + " is busy");
        } else {
            this.currentPiece = manager.getPieceInfo(pieceIndex);
            this.busy = true;
            beginIndex = 0;
            sendRequestMessage(pieceIndex, 0, Constants.BLOCK_SIZE);
        }
    }


    @Override
    public boolean equals(Object obj) {
        if (obj instanceof PeerInstance) {
            return this.connectingPeerId.equals(((PeerInstance) obj).connectingPeerId);
        } else {
            return false;
        }
    }


    @Override
    public int hashCode() {
        return connectingPeerId.hashCode();
    }
}
